package com.zimin.cloud.user.config;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.zimin.cloud.mine.pojo.user.view.UserMomentsVO;
import com.zimin.cloud.user.biz.IUserFollowingBiz;
import com.zimin.cloud.user.biz.impl.UserFollowingBizImpl;
import com.zimin.cloud.user.common.constant.UserMomentsConstant;
import com.zimin.cloud.user.pojo.entity.UserFollowing;
import com.zimin.cloud.user.pojo.entity.UserMoments;
import com.zimin.cloud.user.websocket.WebSocketService;
import com.zimin.common.constant.StringPool;
import com.zimin.common.utils.ObjectUtils;
import com.zimin.common.utils.StringUtils;
import com.zimin.redis.utils.RedisOperator;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;
import java.sql.Array;
import java.util.ArrayList;
import java.util.List;

/**
 * rocket mq
 *
 * @author 恒利
 * @since 2022/6/29 10:00 PM
 */
@Slf4j
@Configuration
public class RocketMQConfig {

    /**
     * rocketmq的服务地址
     */
    @Value("${rocketmq.name.server.address}")
    private String nameServerAddress;

    @Autowired
    private IUserFollowingBiz userFollowingBiz;
    @Autowired
    private RedisOperator redisOperator;

    @Bean("momentsProducer")
    public DefaultMQProducer momentsProducer() throws MQClientException {
        var producer = new DefaultMQProducer(UserMomentsConstant.GROUP_MOMENTS);
        producer.setNamesrvAddr(nameServerAddress);
        producer.start();
        log.info("rocketmq 生产者 创建成功");
        return producer;
    }

    @Bean("momentsConsumer")
    public DefaultMQPushConsumer momentsConsumer() throws MQClientException {
        var consumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_MOMENTS);
        consumer.setNamesrvAddr(nameServerAddress);
        //订阅生产者
        consumer.subscribe(UserMomentsConstant.TOPIC_MOMENTS, StringPool.START);
        /*
         * list : 消息
         * consumeOrderlyContext :上下文
         */
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, consumeConcurrentlyContext) -> {
            var msg = msgs.get(0);
            if (ObjectUtils.isEmpty(msg)) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            var bodyStr = new String(msg.getBody());
            var userMoment = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), UserMomentsVO.class);
            var userId = userMoment.getUserId();
            //获取用户粉丝
            var fanList = userFollowingBiz.getUserFans(userId);
            fanList.forEach(fan -> {
                var key = "subscribed-" + fan.getUserId();
                var subscribedListStr = redisOperator.get(key, String.class);
                List<UserMomentsVO> subscribedList = new ArrayList<>();
                if (StringUtils.isNotEmpty(subscribedListStr)) {
                    subscribedList = JSON.parseArray(subscribedListStr, UserMomentsVO.class);
                }
                subscribedList.add(userMoment);
                redisOperator.set(key, JSONObject.toJSON(subscribedList));
            });
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        log.info("rocketmq consumer 创建成功");
        return consumer;
    }

    @Bean("danmusProducer")
    public DefaultMQProducer damusProducer() throws Exception {
        var producer = new DefaultMQProducer(UserMomentsConstant.DAN_MU_MOMENTS);
        producer.setNamesrvAddr(nameServerAddress);
        producer.start();
        return producer;
    }

    @Bean("danmusConsumer")
    public DefaultMQPushConsumer danmusConsumer() throws Exception {
        //实例化消费者
        var consumer = new DefaultMQPushConsumer(UserMomentsConstant.DAN_MU_MOMENTS);
        //设置NameServer的地址
        consumer.setNamesrvAddr(nameServerAddress);
        //订阅一个或者多个topic，以及tag来过滤需要消费的信息
        consumer.subscribe(UserMomentsConstant.TOPIC_DAN_MU, StringPool.START);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, consumeConcurrentlyContext) -> {
            var msg = msgs.get(0);
            var msgByte = msg.getBody();
            var bodyStr = new String(msgByte);
            var jsonObject = JSONObject.parseObject(bodyStr);
            var sessionId = jsonObject.getString("sessionId");
            var message = jsonObject.getString("message");
            var webSocketService = WebSocketService.WEBSOCKET_MAP.get(sessionId);
            if (webSocketService.getSession().isOpen()) {
               try{
                   webSocketService.sendMessage(message);
               } catch (IOException e){
                   e.printStackTrace();
               }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        log.info("rocketmq consumer 创建成功");
        return consumer;
    }


}
